package com.hliushi.spark.rdd

import org.apache.commons.lang3.StringUtils
import org.apache.spark.{SparkConf, SparkContext}
import org.junit.Test

/**
 * descriptions:
 *
 * author: Hliushi
 * date: 2021/5/13 19:02
 */
class AccessLogAgg {

  // 1.创建SparkContext
  val conf: SparkConf = new SparkConf().setMaster("local[6]").setAppName("AccessLogAgg")
  val sc = new SparkContext(conf)

  /**
   * 给定一个网站的访问记录, 俗称 Access log
   * 计算其中出现的独立ip, 以及其访问的次数, 显示top10
   */
  @Test
  def ipAgg(): Unit = {

    // 2.读取文件, 生成数据集
    // 三种读取方式都相同
    //val ipData = sc.textFile("./dataset/access_log_sample.txt")
    //val ipData = sc.textFile("D:\\Code\\SparkProjects\\spark\\dataset\\access_log_sample.txt")
    val ipData = sc.textFile("dataset/access_log_sample.txt")

    // 3.取出IP, 赋予出现次数为1
    val rdd1 = ipData.map(x => x.split(" ")(0) -> 1)

    // 4.简单清洗
    //    4.1.去除空的数据
    //    4.2.去除非法的数据
    //    4.3.根据业务在规整一下数据

    // 5.根据IP出现的次数进行聚合

    // 6.根据IP出现的次数进行排序
    /**
     * StringUtils.isNotBlank(str)
     * 检查字符串是否 不为空（""），不为null和仅空白。
     * example:
     *  StringUtils.isNotBlank(null)      = false
     *  StringUtils.isNotBlank("")        = false
     *  StringUtils.isNotBlank(" ")       = false
     *  StringUtils.isNotBlank("bob")     = true
     *  StringUtils.isNotBlank("  bob  ") = true
     *
     * return: ("")空, (" ")空白, (null)null   --> false
     */
    val result = rdd1
      .reduceByKey((x, y) => x + y)
      .filter(x => StringUtils.isNotBlank(x._1))
      .sortBy(x => x._2, ascending = false)
      .take(10)


    // 7.取出结果, 打印结果
    result.foreach(println(_))



    // result02.foreach(item => println(item))
  }
}